package net.bwie.realtime.jtp.log.job;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;


/**
 * 简介说明:
 *
 * @author: LiLi
 * @date: 2025/05/23 22:03:22
 * @version: 1.0
 */
public class JtpTrafficAppStartMinuteWindowDwsJobjoin {
    public static void main(String[] args) {
        // 1. 表执行环境
        TableEnvironment tabEnv = getTableEnv();




        // 2. 输入表-input：映射到Kafka消息队列
        createInputTable(tabEnv);

        // 3. 数据处理-select
        handle(tabEnv);

        // 4. 输出表-output：映射到Clickhouse表
        createOutputTable(tabEnv);

        // 5. 保存数据-insert
        saveToClickHouse(tabEnv) ;


        joinDim(tabEnv);
    }

    private static void joinDim(TableEnvironment tabEnv) {

        Table startAdResultTable = tabEnv.sqlQuery(
                "SELECT\n" +
                        "    t1.window_start_time\n" +
                        "    , t1.window_end_time\n" +
                        "    , t2.name AS open_ad_name\n" +
                        "    , t1.open_ad_count\n" +
                        "    , t1.open_ad_ms\n" +
                        "    , t1.open_ad_skip_ms\n" +
                        "    , t1.ts\n" +
                        "FROM start_ad_result_table t1\n" +
                        "    LEFT JOIN ads_info_mysql_source_dim FOR SYSTEM_TIME AS OF t1.proc_time AS t2\n" +
                        "        ON t1.open_ad_id = t2.id"
        );


        tabEnv.createTemporaryView("start_ad_result_table",startAdResultTable);

        tabEnv.executeSql("CREATE TABLE `ads_info_mysql_source_dim`\n" +
                "(\n" +
                "    id               INT COMMENT '主键ID',\n" +
                "    name             STRING COMMENT '广告名称',\n" +
                "    content          STRING COMMENT '广告内容（图片URL/视频URL/H5代码）',\n" +
                "    type             STRING COMMENT '广告类型（IMAGE, VIDEO, HTML5）',\n" +
                "    start_time       TIMESTAMP COMMENT '展示开始时间',\n" +
                "    end_time         DATETIME COMMENT '展示结束时间',\n" +
                "    is_active        INT COMMENT '是否启用',\n" +
                "    target_condition STRING COMMENT '定向条件（JSON格式）',\n" +
                "    created_at       TIMESTAMP COMMENT '数据插入时间',\n" +
                "    updated_at       TIMESTAMP COMMENT '数据更新时间',\n" +
                "    PRIMARY KEY (id) NOT ENFORCED\n" +
                ") WITH (\n" +
                "      'connector' = 'jdbc',\n" +
                "      'url' = 'jdbc:mysql://node103:3306/jtp_ads',\n" +
                "      'table-name' = 'ads_info',\n" +
                "      'lookup.cache.max-rows' = '10',\n" +
                "      'lookup.cache.ttl' = '1 hour',\n" +
                "      'username' = 'root',\n" +
                "      'password' = '123456',\n" +
                "      'driver' = 'com.mysql.cj.jdbc.Driver'\n" +
                "      )");

        tabEnv.sqlQuery("SELECT\n" +
                "    t1.window_start_time\n" +
                "     , t1.window_end_time\n" +
                "     , t2.name AS open_ad_name\n" +
                "     , t1.open_ad_count\n" +
                "     , t1.open_ad_ms\n" +
                "     , t1.open_ad_skip_ms\n" +
                "     , t1.ts\n" +
                "FROM start_ad_result_table t1\n" +
                "         LEFT JOIN ads_info_mysql_source_dim FOR SYSTEM_TIME AS OF t1.proc_time AS t2\n" +
                "ON t1.open_ad_id = t2.id");

    }

    private static void saveToClickHouse(TableEnvironment tabEnv) {
        // 2-保存：APP启动类型统计

    }

    private static void createOutputTable(TableEnvironment tabEnv) {

    }


    private static Table handle(TableEnvironment tabEnv) {
            return null;
    }
    private static void createInputTable(TableEnvironment tabEnv) {

    }

    private static TableEnvironment getTableEnv() {
        // 1环境属性设置
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .useBlinkPlanner()
                .build();
        TableEnvironment tabEnv = TableEnvironment.create(settings);
        // 2配置属性设置
        Configuration configuration = tabEnv.getConfig().getConfiguration();
        configuration.setString("table.local-time-zone", "Asia/Shanghai");
        configuration.setString("table.exec.resource.default-parallelism", "1");
        configuration.setString("table.exec.state.ttl", "5 s");
        // 3返回对象
        return tabEnv;
    }
}
